package com.zhu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhu.bean.TradeUserSpuOrderBean;
import com.zhu.config.KafkaTopicConfig;
import com.zhu.function.DIMAsyncFunction;
import com.zhu.utils.ClickHouseUtil;
import com.zhu.utils.DataFormatUtil;
import com.zhu.utils.TimestampLtz3CompareUtil;
import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 交易域用户-SPU 粒度下单各窗口汇总表
 */

/**
 * WebApp(SpringBoot) Maxwell Kafka Zookeeper HDFS Hbase ClickHouse
 */
public class DWSTradeUseSpuOrderWindowApp {

    public static void main(String[] args) throws Exception {

        //todo env
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(4);  //kafka_topic partition

        //checkPoint
         /*
        streamExecutionEnvironment.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE); //精确一次
        //状态后端
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);  //检查点保存在hdfs
        System.setProperty("HADOOP_USER_NAME", "zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);  //TimeOut
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  //最大共存检查点
      */

        //todo kafka_dwd_order_trade_detail
        String topic = KafkaTopicConfig.KAFKA_DWD_TRADE_ORDER_DETAIL_TOPIC;
        String groupId = "dws_trade_use_spu_order" + KafkaTopicConfig.KAFKA_GROUP_ID_LAST_NAME;
        DataStreamSource<String> kafkaDStream = streamExecutionEnvironment.addSource(ZhuKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //todo json
        SingleOutputStreamOperator<JSONObject> jsonObjDStream = kafkaDStream.flatMap(
                new FlatMapFunction<String, JSONObject>() {
                    @Override
                    public void flatMap(String data, Collector<JSONObject> collector) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(data);
                            collector.collect(jsonObject);
                        } catch (Exception exception) {
                            System.out.println("Dirty >>>");
                            exception.printStackTrace();
                        }
                    }
                }
        );

        //todo 订单明细 id 分组
        KeyedStream<JSONObject, String> keyedBydOrderDetailIdDStream = jsonObjDStream.keyBy(jsonObject -> jsonObject.getString("id"));

        //todo 去重 join 数据 撤回流
        //注册定时器
        SingleOutputStreamOperator<JSONObject> filterDetailDataDStream = keyedBydOrderDetailIdDStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> lastJoinDataState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        ValueStateDescriptor<JSONObject> orderDetailJoinDescriptor = new ValueStateDescriptor<>("last-detail-data", JSONObject.class);
                        lastJoinDataState = getRuntimeContext().getState(orderDetailJoinDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject data, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        //保留最晚来的数据
                        String currentTime = data.getString("row_op_ts");
                        JSONObject stateData = lastJoinDataState.value();
                        if (stateData == null) {
                            lastJoinDataState.update(data);
                            context.timerService().registerProcessingTimeTimer(context.timerService().currentProcessingTime() + 5000L);
                        } else {
                            String lastJoinTime = stateData.getString("row_op_ts");
                            int compareResult = TimestampLtz3CompareUtil.compare(lastJoinTime, currentTime);
                            if (compareResult != 1) {
                                lastJoinDataState.update(data);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        out.collect(lastJoinDataState.value());
                        lastJoinDataState.clear();
                    }
                }
        );

        //todo javabean
        SingleOutputStreamOperator<TradeUserSpuOrderBean> tradeUserSpuDStream = filterDetailDataDStream.map(
                jsonObject -> {
                    HashSet<String> orderIdSet = new HashSet<>();
                    orderIdSet.add(jsonObject.getString("order_id"));
                    return TradeUserSpuOrderBean.builder()
                            .skuId(jsonObject.getString("sku_id"))
                            .userId(jsonObject.getString("user_id"))
                            .orderAmount(jsonObject.getDouble("split_total_amount"))
                            .orderIdSet(orderIdSet)
                            .ts(DataFormatUtil.toTs(jsonObject.getString("create_time"), true))
                            .build();
                }
        );
        //todo  关联sku_info  spu_id tm_id category_3_id
//        tradeUserSpuDStream.map(
//                new RichMapFunction<TradeUserSpuOrderBean, TradeUserSpuOrderBean>() {
//
//                    //建立Phoenix链接
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        super.open(parameters);
//                    }
//
//                    @Override
//                    public TradeUserSpuOrderBean map(TradeUserSpuOrderBean tradeUserSpuOrderBean) throws Exception {
//                        return null;
//                    }
//                }
//        );

     //   tradeUserSpuDStream.print("关联之前>>>>");
        SingleOutputStreamOperator<TradeUserSpuOrderBean> tradeUserSpuWithSkuDStream =
                AsyncDataStream.unorderedWait(tradeUserSpuDStream, new DIMAsyncFunction<TradeUserSpuOrderBean>("DIM_SKU_INFO") {

            @Override
            public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                input.setSpuId(dimInfo.getString("SPU_ID"));
                input.setTrademarkId(dimInfo.getString("TM_ID"));
                input.setCategory3Id(dimInfo.getString("CATEGORY3_ID"));
            }

            @Override
            public String getKey(TradeUserSpuOrderBean input) {
                return input.getSkuId();
            }
        }, 100,TimeUnit.SECONDS);//大于zookeeper连接时间

       // tradeUserSpuWithSkuDStream.print(">>>>>>WITH SKU DATA>>>>>>");

        //todo waterMark
        SingleOutputStreamOperator<TradeUserSpuOrderBean> tradeUserSpuWithWaterMarkDStream = tradeUserSpuWithSkuDStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeUserSpuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<TradeUserSpuOrderBean>() {
                    @Override
                    public long extractTimestamp(TradeUserSpuOrderBean tradeUserSpuOrderBean, long l) {
                        return tradeUserSpuOrderBean.getTs();
                    }
                }));
        //todo  分组 开窗 聚合
        //分组 user_id spu_id  trademark_id  c3_id
        KeyedStream<TradeUserSpuOrderBean, Tuple4<String, String, String, String>> keyedByFourTargetDStream = tradeUserSpuWithWaterMarkDStream.keyBy(new KeySelector<TradeUserSpuOrderBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TradeUserSpuOrderBean tradeUserSpuOrderBean) throws Exception {
                String spuId = tradeUserSpuOrderBean.getSpuId();
                String category3Id = tradeUserSpuOrderBean.getCategory3Id();
                String trademarkId = tradeUserSpuOrderBean.getTrademarkId();
                String userId = tradeUserSpuOrderBean.getSpuId();
                return Tuple4.of(spuId, userId, trademarkId, category3Id);
            }
        });

        WindowedStream<TradeUserSpuOrderBean, Tuple4<String, String, String, String>, TimeWindow> windowTradeUserSpuDStream = keyedByFourTargetDStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithGroupTargetDStream = windowTradeUserSpuDStream.reduce(
                new ReduceFunction<TradeUserSpuOrderBean>() {
                    @Override
                    public TradeUserSpuOrderBean reduce(TradeUserSpuOrderBean value1, TradeUserSpuOrderBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());  //orderCount 定义不了
                        return value1;
                    }

                },
                new WindowFunction<TradeUserSpuOrderBean, TradeUserSpuOrderBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow timeWindow, Iterable<TradeUserSpuOrderBean> iterable, Collector<TradeUserSpuOrderBean> collector) throws Exception {
                        TradeUserSpuOrderBean data = iterable.iterator().next();
                        data.setTs(System.currentTimeMillis());
                        data.setOrderCount((long) data.getOrderIdSet().size());
                        data.setStt(DataFormatUtil.toYmdHms(timeWindow.getStart()));
                        data.setEdt(DataFormatUtil.toYmdHms(timeWindow.getEnd()));
                        collector.collect(data);
                    }
                }
        );

        //todo  关联分组无关维表
        //1 关联spu_info
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithSpuDS = AsyncDataStream.unorderedWait(reduceWithGroupTargetDStream,
                new DIMAsyncFunction<TradeUserSpuOrderBean>("DIM_SPU_INFO") {

                    @Override
                    public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                        input.setSpuName(dimInfo.getString("SPU_NAME"));
                    }

                    @Override
                    public String getKey(TradeUserSpuOrderBean input) {
                        return input.getSpuId();
                    }
                }, 100, TimeUnit.SECONDS);
        //2 关联tm_info 表
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithTMDS = AsyncDataStream.unorderedWait(reduceWithSpuDS,
                new DIMAsyncFunction<TradeUserSpuOrderBean>("DIM_BASE_TRADEMARK") {
                    @Override
                    public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                        input.setTrademarkName("NAME");
                    }

                    @Override
                    public String getKey(TradeUserSpuOrderBean input) {
                      return input.getTrademarkId();
                    }
                },
                100, TimeUnit.SECONDS);
        //3 关联 C3
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithC3DS
                = AsyncDataStream.unorderedWait(reduceWithTMDS, new DIMAsyncFunction<TradeUserSpuOrderBean>("DIM_BASE_CATEGORY3") {
            @Override
            public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                input.setCategory3Name(dimInfo.getString("NAME"));
                input.setCategory2Id(dimInfo.getString("CATEGORY2_ID"));
            }

            @Override
            public String getKey(TradeUserSpuOrderBean input) {
                return input.getCategory3Id();
            }
        }, 100, TimeUnit.SECONDS);
        //4 关联 C2
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithC2DS
                = AsyncDataStream.unorderedWait(reduceWithC3DS, new DIMAsyncFunction<TradeUserSpuOrderBean>("DIM_BASE_CATEGORY2") {
            @Override
            public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                input.setCategory2Name(dimInfo.getString("NAME"));
                input.setCategory1Id(dimInfo.getString("CATEGORY1_ID"));
            }

            @Override
            public String getKey(TradeUserSpuOrderBean input) {
                return input.getCategory2Id();
            }
        }, 100, TimeUnit.SECONDS);
        //5 关联 C1
        SingleOutputStreamOperator<TradeUserSpuOrderBean> reduceWithAllTargetDStream
                = AsyncDataStream.unorderedWait(reduceWithC2DS, new DIMAsyncFunction<TradeUserSpuOrderBean>("DIM_BASE_CATEGORY1") {
                    @Override
                    public void join(TradeUserSpuOrderBean input, JSONObject dimInfo) {
                        input.setCategory1Name(dimInfo.getString("NAME"));
                    }

                    @Override
                    public String getKey(TradeUserSpuOrderBean input) {
                        return input.getCategory1Id();
                    }
                }
                , 100, TimeUnit.SECONDS
        );
        //todo clickHouse
        reduceWithAllTargetDStream.print("join all target >>>>");


        reduceWithAllTargetDStream.addSink(ClickHouseUtil.getClickHouseSinkFunction(
                "insert into dws_trade_user_spu_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
        ));


        //todo execute
        streamExecutionEnvironment.execute("DWSTradeUseSpuOrderWindowApp");



    }

}
